package net.bwie.realtime.dws.log.wea;

import net.bwie.realtime.dws.log.utils.CityCodeConverter;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.deserialization.SimpleListDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;


public class DorisSource {

    public static void main(String[] args) throws  Exception{

        // 1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);

        //2.读取数据源
        DorisOptions option = DorisOptions.builder()
                .setFenodes("node102:8030")
                .setTableIdentifier("user_realtime_report.ads_user_profile_detail")
                .setUsername("root")
                .setPassword("123456")
                .build();

        DorisReadOptions readOptions = DorisReadOptions.builder().build();
        org.apache.doris.flink.source.DorisSource<List<?>> dorisSource = org.apache.doris.flink.source.DorisSource.<List<?>>builder()
                .setDorisOptions(option)
                .setDorisReadOptions(readOptions)
                .setDeserializer(new SimpleListDeserializationSchema())
                .build();

        DataStreamSource<List<?>> source = env.fromSource(dorisSource, WatermarkStrategy.noWatermarks(), "source");

        // 3.将city字段名称转换成cityCode
        SingleOutputStreamOperator<String> map = source.map(new MapFunction<List<?>, String>() {
            @Override
            public String map(List<?> value) throws Exception {
                String userId = value.get(0).toString();
                String city = value.get(2).toString();
                String cityCode = CityCodeConverter.getCityCode(city);
                return userId + "," +city + "," + cityCode;
            }
        });
        map.print();

        //4.请求api数据服务接口


        env.execute("DorisSource");
    }
}
